package jupitermouse.site.source.custom.redis;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
 * https://blog.csdn.net/aa518189/article/details/88726467
 * @desc 自定义redis数据源
 * @author renqiqiang
 */
@Slf4j
public class RedisSourceFunction extends RichSourceFunction<String> {

    private volatile Boolean running = true;

    private transient volatile Integer idState;
    /**
     * 连接池配置
     */
    private JedisPoolConfig config;
    /**
     * 获取连接池
     */
    JedisPool jedisPool =null;
    /**
     * 获取核心对象
     */
    Jedis jedis = null;

    @Override
    public void open(Configuration conf) throws Exception {
        super.open(conf);
        initRedis(conf);

    }

    /**
     * 启动一个 source，即对接一个外部数据源然后 emit 元素形成 stream
     * 大部分情况下会通过在该方法里运行一个 while 循环的形式来产生 stream
     * @param ctx SourceContext
     * @throws Exception Exception
     */
    @Override
    public void run(SourceContext ctx) throws Exception {

        if (idState == null){
            idState = 1;
        }
        while (running){
            Integer nowId = idState;
            //从redis里面自增式得取值，模拟一个流式得过程
            String result = jedis.hget("flink:redis:user", String.valueOf(nowId));
            if (result!=null){
                log.info("id:{}\nrun:{}",nowId, result);
                ctx.collect(result);
                idState= ++nowId;
            }else {
                //如果没有取到，停顿一下
                Thread.sleep(100);
            }
        }
    }

    @Override
    public void cancel() {
        log.info("---------------cancel----------------");
        running = false;
    }

    private void initRedis(Configuration conf){
        String host = conf.getString("flink.redis.host", "127.0.0.1");
        int port = conf.getInteger("flink.redis.port", 6379);
        config = new JedisPoolConfig();
        jedisPool = new JedisPool(config, host, port);
        jedis = jedisPool.getResource();
        log.info("open the redis poll");
    }

}
